package com.zhou.piledriver.config;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.AbstractMqttMessageHandler;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;

import org.springframework.messaging.Message;

import java.util.Objects;

import static cn.hutool.http.ContentType.JSON;

/**
 * @author ZhouShenBin
 * @date 2024/3/20 14:52
 * @description
 */

@Configuration
//@IntegrationComponentScan
@Slf4j
@EnableIntegration
public class MqttConfig {

    @Value("${mqtt.services:#{null}}")
    private String[] mqttServices;//服务器地址以及端口
    @Value("${mqtt.clientId}")
    private String clientId;
    @Value("${mqtt.topic}")
    private String[] topic;
    @Value("${mqtt.userName}")
    private String userName;
    @Value("${mqtt.password}")
    private String password;
    @Value("${mqtt.keepAliveInterval}")
    private int keepAliveInterval;
    @Value("${mqtt.automaticReconnect}")
    private boolean automaticReconnect;
    @Value("${mqtt.completionTimeout}")
    private int completionTimeout;
    @Value("${mqtt.qos}")
    private int qos;

    public final static String INPUT_DATA = "inputData";
    private final static String OUTPUT_DATA = "outputData";
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(mqttServices);
        if(StringUtils.isNotBlank(userName)) {
            options.setUserName(userName);//用户名
        }
        if(null!=password) {
            options.setPassword(password.toCharArray());//密码
        }
        options.setKeepAliveInterval(keepAliveInterval);//心跳时间
        options.setAutomaticReconnect(automaticReconnect);//断开是否自动重联
        factory.setConnectionOptions(options);

        return factory;
    }

    @Bean
    public MessageProducerSupport mqttInput(MqttPahoClientFactory mqttPahoClientFactory){
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(clientId, mqttPahoClientFactory, topic);//建立订阅连接
        DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter();
        converter.setPayloadAsBytes(true);//bytes类型接收
        adapter.setCompletionTimeout(completionTimeout);//连接超时的时间
        adapter.setConverter(converter);
        adapter.setQos(qos);//消息质量
        adapter.setOutputChannelName(INPUT_DATA);//输入管道名称
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = OUTPUT_DATA)
    public AbstractMqttMessageHandler MQTTOutAdapter(MqttPahoClientFactory connectionFactory) {
        //创建一个新的出站管道，由于MQTT的发布与订阅是两个独立的连接，因此客户端的ID(即APPID）不能与订阅时所使用的ID一样，否则在服务端会认为是同一个客户端，而造成连接失败
        MqttPahoMessageHandler outGate = new MqttPahoMessageHandler(clientId + "_put", connectionFactory);
        DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter();
        converter.setPayloadAsBytes(true);//bytes类型接收
        outGate.setAsync(true);
        outGate.setCompletionTimeout(completionTimeout);//设置连接超时时时
        outGate.setDefaultQos(qos);//设置通信质量
        outGate.setConverter(converter);
        return outGate;
    }


}
